package com.atguigu.gmall.realtime.dwd.db.app;

import com.atguigu.edu.realtime.common.base.BaseSQLApp;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdTradeOrderPaySucDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdTradeOrderPaySucDetail().start(
                10478,
                4,
                Constant.DWD_TRANS_PAY
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {


        readOdsDS(tableEnv,Constant.DWD_TRANS_PAY);

        tableEnv.executeSql("create table page_log (\n" +
                "\tcommon map<string,string>\n" +
                /*"  ts bigint,\n" +
                "  et as to_timestamp_ltz(ts, 0) ,\n " +
                "  watermark for et as et - interval '3' second " +*/
                " ) " + SQLUtil.getKafkaDDL(Constant.TOPIC_DWD_TRAFFIC_PAGE, Constant.DWD_TRANS_PAY));

        Table pageLog = tableEnv.sqlQuery("select common['sid'] as sid , common['sc'] as sc from page_log");

        tableEnv.createTemporaryView("sc_log",pageLog);

        Table paymentInfo = tableEnv.sqlQuery("select\n" +
                "    data['id'] id,\n" +
                "    data['order_id'] order_id,\n" +
                "    data['payment_type'] payment_type,\n" +
                "    data['callback_time'] callback_time,\n" +
                "    ts  " +
                "from topic_db\n" +
                "where `database`='edu'\n" +
                "and `table`='payment_info'\n" +
                "and `type`='insert'\n" +
                "and data['payment_status'] = '1602'");

        tableEnv.createTemporaryView("payment_info",paymentInfo);

        //tableEnv.executeSql("select * from payment_info").print();


        Table orderDetail = tableEnv.sqlQuery("select\n" +
                "    data['id'] id,\n" +
                "    data['order_id'] order_id,\n" +
                "    data['user_id'] user_id,\n" +
                "    data['course_id'] course_id,\n" +
                "    data['course_name'] course_name,\n" +
                "    data['origin_amount'] origin_amount,\n" +
                "    data['coupon_reduce'] coupon_reduce,\n" +
                "    data['final_amount'] final_amount\n" +
                "from topic_db\n" +
                "where `database`='edu'\n" +
                "and `table`='order_detail'\n" +
                "and `type`='insert'");

        tableEnv.createTemporaryView("order_detail",orderDetail);

        //tableEnv.executeSql("select * from order_detail").print();


        Table orderInfo = tableEnv.sqlQuery("select\n" +
                "    data['id'] order_id,\n" +
                "    data['province_id'] province_id,\n" +
                "    data['session_id'] session_id,\n" +
                "    data['order_status'] order_status\n" +
                "from topic_db\n" +
                "where `database`='edu'\n" +
                "and `table`='order_info'\n" +
                "and `type`='insert'");

        tableEnv.createTemporaryView("order_info",orderInfo);

        //tableEnv.executeSql("select * from order_info").print();

        Table result = tableEnv.sqlQuery(
                "select\n" +
                        "\tpi.id,\n" +
                        "\tpi.order_id,\n" +
                        "\tpi.payment_type,\n" +
                        "\tpi.callback_time,\n" +
                        "\tod.user_id,\n" +
                        "\tod.course_id,\n" +
                        "\tod.course_name,\n" +
                        "\tod.origin_amount,\n" +
                        "\tod.coupon_reduce,\n" +
                        "\tod.final_amount,\n" +
                        "\toi.province_id,\n" +
                        "\toi.order_status,\n" +
                        "    oi.session_id,\n" +
                        "    sl.sc,\n" +
                        " ts "+
                        "from payment_info pi\n" +
                        " join order_detail od\n" +
                        "on pi.order_id = od.order_id\n" +
                        " join order_info oi \n" +
                        "on pi.order_id = oi.order_id\n"+
                        " join sc_log sl\n" +
                        "on oi.session_id = sl.sid"
        );

        //result.execute().print();

       tableEnv.executeSql("create table "+Constant.DWD_TRANS_PAY+"(\n" +
                "\tid string,\n" +
                "\torder_id string,\n" +
                "\tpayment_type string,\n" +
                "\tcallback_time string,\n" +
                "\tuser_id string,\n" +
                "\tcourse_id string,\n" +
                "\tcourse_name string,\n" +
                "\torigin_amount string,\n" +
                "\tcoupon_reduce string,\n" +
                "\tfinal_amount string,\n" +
                "\tprovince_id string,\n" +
                "\torder_status string,\n" +
                "\tsession_id string,\n" +
                "\tsc string,\n" +
                "\tts bigint,\n" +
                "PRIMARY KEY (id) NOT ENFORCED  " +
                " ) " + SQLUtil.getUpsertKafkaDDL(Constant.DWD_TRANS_PAY));

        result.executeInsert(Constant.DWD_TRANS_PAY);


    }
}
